package cn.tedu.csmall.cart.webapi.kafka;

import cn.tedu.csmall.commons.pojo.cart.entity.Cart;
import com.google.gson.Gson;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

// 当前Consumer是用于接收kafka消息的类
// 要求将这个类也保存到Spring容器中,因为SpringKafka框架使用Spring容器中的对象
@Component
public class Consumer {

    // Spring-Kafka接收消息,使用了"监听机制"
    // 框架设计了一条线程,实时关注Kafka话题接收的情况
    // 我们可以指定一个话题名称(myCart),设置一旦这个话题中有消息,监听线程就通知下面方法运行
    @KafkaListener(topics = Producer.TOPIC_KEY)
    // 上面注解就实现了监听的机制
    // 当前Kafka的myCart话题出现消息时,会自动运行下面方法
    // 下面方法的参数和返回值是不能修改的(方法名随意)
    public void received(ConsumerRecord<String,String> record){
        // 方法的返回值必须是void ,参数必须是ConsumerRecord
        // ConsumerRecord<[话题类型],[消息类型]>
        // 从record对象中获取消息内容
        String json=record.value();
        // json可能是这样的字符串:{"id":"1","userId":"UU100","price":"10",...}
        // 最终要活的Cart对象,需要将json字符串转换为java对象
        // 仍然使用Gson工具类实现
        Gson gson=new Gson();
        Cart cart=gson.fromJson(json, Cart.class);
        // 转换完成输出cart,测试验证消息的接收效果
        System.out.println(cart);

    }




}
